// Tencent is pleased to support the open source community by making
// 蓝鲸智云 - 监控平台/日志平台 (BlueKing - Monitor/Log) available.
// Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
// Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://opensource.org/licenses/MIT
// Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.
//

package receiver

import (
	"crypto/tls"
	"errors"
	"net"
	"net/http"
	"time"

	"github.com/TencentBlueKing/bkmonitor-kits/logger"
	"github.com/elastic/beats/libbeat/common/transport/tlscommon"
	"github.com/elastic/beats/libbeat/outputs/transport"
	"google.golang.org/grpc"

	"github.com/TencentBlueKing/bk-collector/confengine"
	"github.com/TencentBlueKing/bk-collector/define"
	"github.com/TencentBlueKing/bk-collector/internal/middleware"
)

const (
	ConfigFieldReceiver = "receiver"
)

type Receiver struct {
	httpSrv     *http.Server
	httpTls     *transport.TLSConfig
	benchServer *BenchmarkServer

	ch      CacheHolder
	grpcSrv *grpc.Server
	config  Config
}

var (
	globalRecords     = define.NewRecordQueue(define.PushModeGuarantee)
	globalGrpcSrv     = grpc.NewServer()
	globalCacheHolder CacheHolder
)

// Records 返回 Receiver 全局消息管道
func Records() <-chan *define.Record {
	return globalRecords.Get()
}

// PublishRecord 将数据推送只 Receiver 全局消息管道
func PublishRecord(r *define.Record) {
	globalRecords.Push(r)
}

// GetCacheHolder 获取全局 CacheHolder
func GetCacheHolder() CacheHolder {
	return globalCacheHolder
}

func New(conf *confengine.Config) (*Receiver, error) {
	var c Config
	var err error

	if err = conf.UnpackChild(ConfigFieldReceiver, &c); err != nil {
		return nil, err
	}
	logger.Infof("Receiver config: %+v", c)

	var tlsConfig *tlscommon.TLSConfig
	if c.HttpServer.TLS != nil {
		if tlsConfig, err = tlscommon.LoadTLSServerConfig(c.HttpServer.TLS); err != nil {
			return nil, err
		}
		logger.Infof("Receiver start httpserver with tls config: %+v", tlsConfig)
	}

	globalCacheHolder, err = NewOneApmCacheHolder(conf)
	if err != nil {
		return nil, err
	}

	return &Receiver{
		config:      c,
		httpTls:     tlsConfig,
		httpSrv:     &http.Server{Handler: HttpRouter()},
		grpcSrv:     globalGrpcSrv,
		benchServer: NewBenchmarkServer(c),
	}, nil
}

func (r *Receiver) startBenchServer() error {
	if !r.config.BenchmarkServer.Enabled {
		return nil
	}

	return r.benchServer.Start()
}

func (r *Receiver) startHttpServer() error {
	if !r.config.HttpServer.Enabled {
		return nil
	}

	for _, mid := range r.config.HttpServer.Middlewares {
		handleMiddleware := middleware.GetMiddleware(mid)
		if handleMiddleware != nil {
			r.httpSrv.Handler = handleMiddleware(r.httpSrv.Handler)
		}
	}

	endpoint := r.config.HttpServer.Endpoint
	logger.Infof("start to listen http server at: %v", endpoint)
	if r.httpTls != nil {
		c := r.httpTls.BuildModuleConfig(endpoint)
		l, err := tls.Listen("tcp", endpoint, c)
		if err != nil {
			return err
		}
		return r.httpSrv.Serve(l)
	}

	l, err := net.Listen("tcp", endpoint)
	if err != nil {
		return err
	}

	logger.Infof("register http route: %+v", HttpRoutes())
	logger.Infof("register otlp encoder: %+v", OtlpEncoders())
	return r.httpSrv.Serve(l)
}

func (r *Receiver) startGrpcServer() error {
	if !r.config.GrpcServer.Enabled {
		return nil
	}
	endpoint := r.config.GrpcServer.Endpoint
	logger.Infof("start to listen grpc server at: %v", endpoint)

	l, err := net.Listen(r.config.GrpcServer.Transport, endpoint)
	if err != nil {
		return err
	}
	return r.grpcSrv.Serve(l)
}

func (r *Receiver) Start() error {
	logger.Info("Receiver start working...")

	var errs []error
	go func() {
		if err := r.startHttpServer(); err != nil {
			if errors.Is(err, http.ErrServerClosed) {
				logger.Info("Receiver http server stopped")
				return
			}
			errs = append(errs, err)
		}
	}()
	go func() {
		if err := r.startGrpcServer(); err != nil {
			errs = append(errs, err)
		}
	}()
	go func() {
		if err := r.startBenchServer(); err != nil {
			errs = append(errs, err)
		}
	}()

	// 确保 goroutine 能被调度
	time.Sleep(time.Millisecond * 100)

	if len(errs) > 0 {
		return errs[0]
	}
	return nil
}

func (r *Receiver) Stop() error {
	if err := r.httpSrv.Close(); err != nil {
		return err
	}
	r.grpcSrv.Stop()
	return nil
}

func (r *Receiver) Reload(conf *confengine.Config) error {
	ch, err := NewOneApmCacheHolder(conf)
	if err != nil {
		return err
	}

	globalCacheHolder = ch
	return nil
}
